package com.pie4cloud.pie.bus.api.consumer.factory;

import com.pie4cloud.pie.bus.api.consumer.IKfkaConsumer;
import com.pie4cloud.pie.bus.api.consumer.annotation.KfkConsumer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class IKfkFactory implements ApplicationContextAware, DisposableBean {
	private  ConcurrentHashMap<String, IKfkaConsumer> services=new ConcurrentHashMap<>();

	@Override
	public void destroy() throws Exception {

	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		//ConcurrentHashMap<String, IKfkaConsumer> services=new ConcurrentHashMap<>();
		Map<String, Object> serviceMap = applicationContext.getBeansWithAnnotation(KfkConsumer.class);
		if (serviceMap!=null && serviceMap.size()>0) {
			for (Object serviceBean : serviceMap.values()) {
				if (serviceBean instanceof IKfkaConsumer) {
					IKfkaConsumer iKfkaConsumer= (IKfkaConsumer) serviceBean;
					KfkConsumer annotation = iKfkaConsumer.getClass().getAnnotation(KfkConsumer.class);
					services.put(annotation.handler(),iKfkaConsumer);
				}
			}
		}
		services.putAll(services);
	}

	public IKfkaConsumer get(String serviceName){
		return services.get(serviceName);
	}
}
